package com.xinqi.common.rocketmq.util;

import com.xinqi.common.rocketmq.dto.RocketMqBean;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.SessionCredentialsProvider;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.MessageListener;

import java.time.Duration;
import java.util.Collections;

/**
 * 消费者帮助
 *
 * @author linkai
 * @date 2023/11/19
 */
public final class MqHelpUtils {
    public static void createPushConsumer(String topic, String groupName, String tag,
                                          int consumptionThreadCount,
                                          RocketMqBean rocketMqBean,
                                          MessageListener listener) throws ClientException {
        ClientServiceProvider provider = provider();
        ClientConfiguration clientConfiguration = clientConfiguration(
            rocketMqBean.getEndpoints(),
            rocketMqBean.getUsername(),
            rocketMqBean.getPassword()
        );

        //订阅消息的过滤规则，表示订阅所有Tag的消息。
        FilterExpression filterExpression;
        //订阅消息的过滤规则，表示订阅所有Tag的消息。
        if (StringUtils.isNotBlank(tag)) {
            filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        } else {
            filterExpression = new FilterExpression();
        }
        //初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        provider.newPushConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            //设置并行消费线程数
            .setConsumptionThreadCount(consumptionThreadCount > 0 ? consumptionThreadCount : 5)
            //设置消费者分组。
            .setConsumerGroup(groupName)
            //设置预绑定的订阅关系。
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            //设置消费监听器。
            .setMessageListener(listener)
            .build();
    }

    public static ClientServiceProvider provider() {
        return ClientServiceProvider.loadService();
    }

    public static ClientConfiguration clientConfiguration(String endPoints, String username, String password) {
        SessionCredentialsProvider sessionCredentialsProvider = new StaticSessionCredentialsProvider(username, password);
        return ClientConfiguration.newBuilder()
            .setEndpoints(endPoints)
            .setCredentialProvider(sessionCredentialsProvider)
            .setRequestTimeout(Duration.ofSeconds(10))
            .build();
    }
}
